#include "ws_server.hpp"
#include "container.hpp"

#include "wlog.hpp"
#include <json.hpp>

using namespace std;

namespace bz {

	

	void ws_manage::on_open(coro_http_connection* conn)
	{
		all_client_set_.insert(conn);
	}

	void ws_manage::on_close(coro_http_connection* conn)
	{
		WLOG << " ws on_close";
		
		all_client_set_.erase(conn);

		for (auto it: romm_ws_map_)
		{
			it.second->conn_set_.erase(conn);
		}

		WLOG << "all_client_set_.size = " << all_client_set_.size();

		for (auto it : romm_ws_map_)
		{
			WLOG << "romm_ws_map_ set  size = " << it.second->conn_set_.size();
		}

	}

	void ws_manage::broad_meet_msg(int& meet_id, std::string msg)
	{
		auto rs = romm_ws_map_.find(meet_id);
		if (rs != romm_ws_map_.end()) {
			for (auto it: rs->second->conn_set_)
			{
				
				//auto rc = async_simple::coro::syncAwait(it->write_websocket(msg));
				////write_msg(it, msg);
				//if (rc) {
				//	on_close(it);
				//}
				coro_http_connection* conn = it;
				
				//WLOG << "broad_meet_msg= " << msg << "conn="<<conn->remote_address();
				
				write_msg(conn, msg).via(&conn->get_executor()).detach();
				
				
				
			}
		}
	}

	void ws_manage::broad_msg(std::string msg)
	{
		for (auto it : all_client_set_)
		{
			//async_simple::coro::syncAwait(it->write_websocket(msg));
			//write_msg(it,msg);
		}
	}

	async_simple::coro::Lazy<void> ws_manage::on_message(coro_http_connection* conn ,std::string_view& data)
	{
		std::string str = std::string(data);

		try
		{
			auto req_body = json::parse(str);
			string type = req_body["type"];
			

			json res_json = {
					{"type","subscrible"},
					{"data",json::object()},
					{"msg",""}
			};


			if (type == "subscrible")
			{
				int meet_id = req_body["meet_id"];
				auto it = romm_ws_map_.find(meet_id);
				if (it != romm_ws_map_.end())
				{
					it->second->conn_set_.insert(conn);
				}
				else
				{
					room_conn* room_sub = new room_conn();
					room_sub->conn_set_.insert(conn);
					romm_ws_map_[meet_id] = room_sub;
				}
				co_await conn->write_websocket(res_json.dump());

			}
			else if (type == "cancel_subscrible") {
				int meet_id = req_body["meet_id"];
				auto it = romm_ws_map_.find(meet_id);
				if (it != romm_ws_map_.end())
				{
					it->second->conn_set_.erase(conn);
				}
				
				co_await conn->write_websocket(res_json.dump());

			}
			else if (type == "record_log") {
				json log_body = req_body["log_body"];
				FLOG << log_body.dump();
			}


			


		}
		catch (const std::exception&)
		{

		}

		co_return;
	}

	async_simple::coro::Lazy<void> ws_manage::write_msg(coro_http_connection* conn, std::string msg)
	{
		auto rc = co_await conn->write_websocket(msg);
		if (rc) {
			on_close(conn);
		}
		co_return;
	}

	ws_manage::ws_manage()
	{

	}

	async_simple::coro::Lazy<void> ws_server::hand(coro_http_request& req, coro_http_response& resp)
	{
		
		if (req.get_content_type() == content_type::websocket) {
			WLOG << "ws_point=" << this;
			coro_http_connection* conn = req.get_conn();
			ws_manage::instance()->on_open(conn);
			websocket_result result{};
			while (true) {
				result = co_await conn->read_websocket();
				
				if (result.ec) {
					ws_manage::instance()->on_close(conn);
					co_return;
				}
				co_await ws_manage::instance()->on_message(conn, result.data);
			}
		}
	}

	
}